package com.lagou.client;

import com.lagou.service.UserService;
import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheListener;
import org.apache.curator.retry.ExponentialBackoffRetry;

import java.util.*;
import java.util.concurrent.ConcurrentHashMap;


public class ClientBootStrap {

    private static  final Map<String,RpcConsumer> rpcConsumerMap = new ConcurrentHashMap<>();

    public static void main(String[] args) throws Exception {
        RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3);
        CuratorFramework client = CuratorFrameworkFactory.builder()
                .connectString("127.0.0.1"+":"+2181)
                .sessionTimeoutMs(5000)
                .connectionTimeoutMs(3000)
                .retryPolicy(retryPolicy)
                .build();
        client.start();

        List<String> services = client.getChildren().forPath("/rpc");
        for (String service : services) {
            String pPath = "/rpc/" + service + "/providers";
            List<String> hosts = client.getChildren().forPath(pPath);
            PathChildrenCache pathChildrenCache = new PathChildrenCache(client,pPath,true);
            pathChildrenCache.getListenable().addListener(pathChildrenCacheListener());
            pathChildrenCache.start(PathChildrenCache.StartMode.BUILD_INITIAL_CACHE);
            RpcConsumer rpcConsumer = new RpcConsumer();
            rpcConsumer.setClient(client);
            rpcConsumer.setpPath(pPath);
            rpcConsumer.initClient(hosts);
            rpcConsumerMap.put(pPath,rpcConsumer);
        }

        new Timer().schedule(new TimerTask() {
            @Override
            public void run() {
                for (Map.Entry<String, RpcConsumer> entry : rpcConsumerMap.entrySet()) {
                    RpcConsumer value = entry.getValue();
                    try {
                        value.reset();
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                }
            }
        }, new Date(),5000);

        RpcConsumer rpcConsumer = rpcConsumerMap.get("/rpc/" + UserService.class.getName() + "/providers");
        UserService proxy = (UserService) rpcConsumer.createProxy(UserService.class);

        while (true){
            Thread.sleep(6000);
            proxy.sayHello("are you ok?");
            System.out.println("已响应");
        }

    }

    public static PathChildrenCacheListener pathChildrenCacheListener(){
        return (client, event) -> {
            String path = event.getData().getPath();
            System.out.println(path);
            int index = path.lastIndexOf("/");
            String service = path.substring(0, index);
            String host = path.substring(index + 1);
            switch (event.getType()){
                case CHILD_REMOVED:
                    rpcConsumerMap.get(service).close(host);
                    return;
                case CHILD_ADDED:
                    rpcConsumerMap.get(service).initClient(host);
                    return;
                case CHILD_UPDATED:
                    rpcConsumerMap.get(service).initClient(host);
                default:
                    System.out.println("事件类型："  + event.getType() + "；操作节点：" + event.getData().getPath());
            }
        };
    }

}